package main

import (
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"

	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
	// Kafka broker 地址
	//brokers := "192.168.3.64:30092" // 替换为 Kafka broker 的实际 IP 地址和端口号

	// 要消费的主题
	topic := "dosec_asset_change" // 替换为你的主题名称

	// 配置消费者
	config := &kafka.ConfigMap{
		"bootstrap.servers":     "192.168.3.134:30092",
		"group.id":              "assets-dataa-engine", // 消费者组ID
		"auto.offset.reset":     "earliest",            // 从最早的消息开始消费
		"security.protocol":     "SASL_PLAINTEXT",      // 使用 SASL 认证
		"sasl.mechanism":        "PLAIN",               // SASL 机制（如 PLAIN、SCRAM-SHA-256 等）
		"sasl.username":         "admin",               // SASL 用户名
		"sasl.password":         "36UdcuHtpZSfRj",      // SASL 密码
		"broker.address.family": "v4",
	}

	// 创建消费者实例
	consumer, err := kafka.NewConsumer(config)
	if err != nil {
		log.Fatalf("无法创建消费者: %v", err)
	}
	defer consumer.Close()

	// 订阅主题
	err = consumer.Subscribe(topic, nil)
	if err != nil {
		log.Fatalf("无法订阅主题: %v", err)
	}

	// 处理系统信号以优雅关闭消费者
	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

	// 消费消息
	run := true
	for run {
		select {
		case sig := <-sigchan:
			fmt.Printf("接收到信号 %v，正在关闭消费者...\n", sig)
			run = false
		default:
			// 拉取消息
			ev := consumer.Poll(100) // 超时时间为100毫秒
			if ev == nil {
				continue
			}

			switch e := ev.(type) {
			case *kafka.Message:
				// 打印消息内容
				fmt.Printf("接收到消息: 主题=%s 分区=%d 偏移量=%d 键=%s 值=%s\n",
					*e.TopicPartition.Topic, e.TopicPartition.Partition, e.TopicPartition.Offset,
					string(e.Key), string(e.Value))
				// 如果只需要消费一条消息，可以在这里退出
				run = false
			case kafka.Error:
				// 处理错误
				if e.Code() == kafka.ErrTimedOut {
					continue
				}
				log.Printf("消费者错误: %v\n", e)
				run = false
			default:
				// 其他事件类型可以在这里处理
				// log.Printf("忽略事件: %v\n", e)
			}
		}
	}

	fmt.Println("消费者已关闭。")
}
